package com.batchengine.pool;

import com.batchengine.BatchEngineException;
import static com.batchengine.util.LoggingFactory.createLogger;

import com.batchengine.worker.DataConsumer;
import com.batchengine.worker.DataGatherer;
import com.batchengine.worker.Worker;
import org.apache.log4j.Logger;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by sunil.incubator@yahoo.com.
 * User: sunil.kalva
 * Date: Jul 11, 2009
 */
public class ProducerConsumerWorkerPool extends WorkerPool {

    private static final Logger logger = createLogger();

    private Long taskId;
    private DataGatherer dataGatherer;
    private boolean poolInitialized = false;

    protected ProducerConsumerWorkerPool(int minWorkers, int maxWorkers, DataGatherer dataGatherer, WorkerFactory workerFactory) {
        super(minWorkers, maxWorkers, workerFactory);
        this.dataGatherer = dataGatherer;
    }

    protected ProducerConsumerWorkerPool(String name, int minWorkers, int maxWorkers, DataGatherer dataGatherer, WorkerFactory workerFactory) {
        super(name, minWorkers, maxWorkers, workerFactory);
        this.dataGatherer = dataGatherer;
    }

    protected ProducerConsumerWorkerPool(String name, DataGatherer dataGatherer, WorkerFactory workerFactory) {
        super(name, workerFactory);
        this.dataGatherer = dataGatherer;
    }

    protected ProducerConsumerWorkerPool(DataGatherer dataGatherer, WorkerFactory workerFactory) {
        super(workerFactory);
        this.dataGatherer = dataGatherer;
    }

    protected ProducerConsumerWorkerPool(int minWorkers, int maxWorkers, DataGatherer dataGatherer) {
        super(minWorkers, maxWorkers);
        this.dataGatherer = dataGatherer;
    }

    protected ProducerConsumerWorkerPool(int minWorkers, int maxWorkers, String name, DataGatherer dataGatherer) {
        super(minWorkers, maxWorkers, name);
        this.dataGatherer = dataGatherer;
    }

    public Long getTaskId() {
        return taskId;
    }

    public void setTaskId(Long taskId) {
        this.taskId = taskId;
    }

    public DataGatherer getDataGatherer() {
        return dataGatherer;
    }

    public synchronized void startWorkers() {

        if (poolInitialized) {
            throw new BatchEngineException("Pool already started");
        }
        startDataGatherer();
        startDataConsumers();

        if (null != getSignal()) {
            logger.debug("@@@@@@@@ starting signal @@@@@@@@");
            getSignal().start();
        }
        poolInitialized = true;
    }

    protected synchronized boolean stopWorkers(int numWorkersToKill) {

        if (super.stopWorkers(numWorkersToKill)) {
            if (null != dataGatherer) {
                dataGatherer.stop();
            }
            poolInitialized = false;
            return true;
        }
        return false;
    }

    private void startDataGatherer() {

        if (null == dataGatherer) {
            throw new BatchEngineException("DataGatherer cant be null in ProducerConsumerWorkerPool");
        }

        Lock lock = new ReentrantLock();
        try {
            lock.lock();
            dataGatherer.start();
        } catch (Exception e) {
            throw new BatchEngineException("DataGatherer not initialized or null ", e);
        } finally {
            lock.unlock();
        }
    }

    private void startDataConsumers() {

        //start the min number of instant threads.
        for (int i = 1; i <= minWorkers; i++) {
            DataConsumer worker = (DataConsumer) addWorker();
            logger.debug(String.format("worker %s  created", worker));
        }
    }

    protected Worker addWorker() {

        DataConsumer worker = null;
        Lock lock = new ReentrantLock();
        try {

            lock.lock();
            String threadName = String.format("%s::%s", this.name, this.nextThreadNumber++);
            worker = (DataConsumer) createWorker(threadName);
            worker.setDataGatherer(dataGatherer);
            worker.start();
            workers.put(threadName, worker);
        } catch (Exception e) {
            throw new BatchEngineException("Worker not initialized or null ", e);
        } finally {
            lock.unlock();
        }
        return worker;
    }

    protected boolean reInit(DataGatherer dataGatherer) {
        stopWorkers();
        this.dataGatherer = dataGatherer;
        return true;
    }
}